package com.relytosoft.mizar.media.camera.services.consumer;

import cn.hutool.core.lang.UUID;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.relytosoft.mizar.media.camera.api.entity.CameraBase;
import com.relytosoft.mizar.media.camera.api.entity.CameraChannel;
import com.relytosoft.mizar.media.camera.api.entity.CameraPlatform;
import com.relytosoft.mizar.media.camera.services.constant.EncodeDeviceProtocol;
import com.relytosoft.mizar.media.camera.services.service.HkSyncService;
import com.relytosoft.mizar.media.common.utils.AesUtils;
import com.relytosoft.mizar.media.external.api.camera.request.PageInfo;
import com.relytosoft.mizar.media.external.api.camera.response.ChannelDto;
import com.relytosoft.mizar.media.external.api.camera.response.RegionDeviceDto;
import com.relytosoft.mizar.media.external.api.camera.response.Root;
import com.relytosoft.mizar.media.external.api.camera.service.CameraService;
import com.relytosoft.mizar.media.external.api.camera.service.ChannelService;
import com.relytosoft.mizar.media.system.api.pojo.dto.DistrictFeignDto;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author lj
 * @date 2022/4/14 13:38
 */
@Component
public class CameraConsumer {
    protected Logger log = LoggerFactory.getLogger(this.getClass());

//    @Resource
//    private KafkaProducer kafkaProducer;

    @Resource
    private CameraService cameraService;

    @Resource
    private ChannelService channelService;

    @Resource
    private HkSyncService hkSyncService;

//    @KafkaListener(id = "consumer-media-camera-sync", topics = MediaMqConstants.CAMERA_TOPIC, containerFactory = "singleContainerFactory")
//    public void processSingleMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) {
//        try {
//            boolean isConsume = false;
//            KafkaOpMsg kafkaOpMsg = JSON.parseObject(record.value().toString(), KafkaOpMsg.class);
//            switch (kafkaOpMsg.getOpType()) {
//                case "SELECT_HK_ENCODE":
//                    isConsume = selectHkEncode(kafkaOpMsg.getData());
//                    break;
//                case "INSERT_OWN_CAMERA":
//                    isConsume = insertOwnCamera(kafkaOpMsg.getData());
//                    break;
//                case "SELECT_HK_CHANNEL":
//                    isConsume = selectHkChannel();
//                    break;
//                case "INSERT_OWN_CHANNEL":
//                    isConsume = insertOwnChannel(kafkaOpMsg.getData());
//                default:
//            }
//            log.info("线程:{} 消费->{}", Thread.currentThread(), kafkaOpMsg.getOpType());
//            if (isConsume) {
//                ack.acknowledge();
//            }
//        } catch (Exception e) {
//            e.printStackTrace();
//            log.error(e.getMessage());
//        }
//    }


    private boolean insertOwnChannel(String data) {
        List<ChannelDto> channelDtoList = JSON.parseObject(data, new TypeReference<List<ChannelDto>>() {
        });
        List<String> channelIndexCodes = channelDtoList.stream().map(ChannelDto::getCameraIndexCode).collect(Collectors.toList());

        List<CameraPlatform> selectList = hkSyncService.selectAllPlatformCode(2, 1);
        if (!selectList.isEmpty()) {
            List<String> ownChannelIndexCodes = selectList.stream().map(CameraPlatform::getPlatformChannelCode).collect(Collectors.toList());
            channelIndexCodes.removeAll(ownChannelIndexCodes);
        }
        if (channelIndexCodes.size() > 0) {
            //收集所有需要添加的数据
            List<ChannelDto> needAddData = channelDtoList
                    .stream()
                    .filter(item -> channelIndexCodes.contains(item.getCameraIndexCode()))
                    .collect(Collectors.toList());
            List<CameraChannel> cameraChannelList = convertCameraChannel(needAddData);
            //获取设备id
            Map<String, Integer> map = selectList
                    .stream()
                    .collect(Collectors.toMap(CameraPlatform::getPlatformCameraCode, CameraPlatform::getOwnId));

            //设置设备id
            cameraChannelList = cameraChannelList.stream()
                    .peek(item -> item.setCameraId(map.get(item.getEncodeDevIndexCode())))
                    .collect(Collectors.toList());

            //过滤没有设备id得通道。没有设备就没有通道
            List<CameraChannel> channelList = cameraChannelList.stream()
                    .filter(cameraChannel -> cameraChannel.getCameraId() != null)
                    .collect(Collectors.toList());

            //批量添加通道信息
            boolean isConsume = hkSyncService.batchInsertCameraChannel(channelList);
            //关联表数据添加
            if (isConsume) {
                List<CameraPlatform> cameraPlatformList = convertToCameraPlatform(channelList);
                isConsume = hkSyncService.batchInsertCameraPlatform(cameraPlatformList);
                if (isConsume) {
                    //更新设备表设备类型
                    //1. 查询
                    Set<Integer> cameraIds = channelList.stream().map(CameraChannel::getCameraId).collect(Collectors.toSet());
                    Map<Integer, Integer> cameraIdType = channelList.stream().collect(Collectors.toMap(CameraChannel::getCameraId, CameraChannel::getCameraTypeId));
                    List<CameraBase> cameraBaseList = hkSyncService.batchSelectByCameraIds(cameraIds);
                    cameraBaseList.forEach(item -> item.setCameraTypeId(cameraIdType.get(item.getCameraTypeId())));
                    //更新
                    isConsume = hkSyncService.batchUpdateCameraBase(cameraBaseList);
                }
            }
            return isConsume;
        }
        return true;
    }

    private List<CameraChannel> convertCameraChannel(List<ChannelDto> channelDtoList) {
        //拼接通道数据对象
        List<CameraChannel> cameraChannelList = new ArrayList<>();
        for (ChannelDto channelDTO : channelDtoList) {
            CameraChannel channel = new CameraChannel();
            //通道编码
            channel.setChannelCode(UUID.fastUUID().toString());
            //通道名称
            channel.setChannelName(channelDTO.getCameraName());
            //连接协议
            channel.setTransType(channelDTO.getTransType() == 1 ? 1000 : 1001);
            //摄像机类型
//            channel.setCameraTypeId(cameraTypeService.findByName(channelDTO.getCameraTypeName()).getId());
            //设备能力集
            channel.setCamability(channelDTO.getCapabilitySet());
            //经度
            channel.setLongItude(channelDTO.getLongitude());
            //纬度
            channel.setLatItude(channelDTO.getLatitude());
            //状态
            channel.setStatus(1);
            //通道编号
            channel.setChannelNum(channelDTO.getChannelNo());
            //编码设备唯一码
            channel.setEncodeDevIndexCode(channelDTO.getEncodeDevIndexCode());
            //通道编码
            channel.setPlatformChannelCode(channelDTO.getCameraIndexCode());
            //添加到集合
            cameraChannelList.add(channel);
        }
        return cameraChannelList;
    }

    /**
     * 查询海康数据
     *
     * @return 确认消息是否接收成功
     */
    private boolean selectHkChannel() {
        //获取海康返回数据
        Root<ChannelDto> root = channelService.channelPage(new PageInfo(1L, 1000L));
//        KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
//        kafkaOpMsg.setOpType("INSERT_OWN_CHANNEL");
//        kafkaOpMsg.setData(JSON.toJSONString(root.getList()));
//        kafkaProducer.send(MediaMqConstants.CAMERA_TOPIC, JSON.toJSONString(kafkaOpMsg), true);
        return true;
    }

    private boolean insertOwnCamera(String data) {
        Map<String, String> objectMap = JSONObject.parseObject(data, new TypeReference<Map<String, String>>() {
        });

        List<RegionDeviceDto> hkCameraData = JSONObject.parseObject(objectMap.get("hkData"), new TypeReference<List<RegionDeviceDto>>() {
        });

        List<DistrictFeignDto> districtFeignDTOList = JSONObject.parseObject(objectMap.get("area"), new TypeReference<List<DistrictFeignDto>>() {
        });
        //获取海康设备唯一编码
        List<String> indexCodes = hkCameraData
                .stream()
                .map(RegionDeviceDto::getIndexCode)
                .collect(Collectors.toList());
        //获取本系统已经存储的三方平台编码并去重
        List<CameraPlatform> selectList = hkSyncService.selectAllPlatformCode(1, 1);
        if (!selectList.isEmpty()) {
            List<String> ownIndexCodes = selectList.stream().map(CameraPlatform::getPlatformCameraCode).collect(Collectors.toList());
            //求差集
            indexCodes.removeAll(ownIndexCodes);
        }

        //添加数据
        if (indexCodes.size() > 0) {
            //1. 收集所有需要添加的数据
            List<RegionDeviceDto> regionDeviceDtoList = hkCameraData
                    .stream()
                    .filter(item -> indexCodes.contains(item.getIndexCode()))
                    .collect(Collectors.toList());
            //2. 收集的数据转为自己数据库的数据
            List<CameraBase> cameraBaseList = convertData(regionDeviceDtoList, districtFeignDTOList);
            boolean isSuccess = hkSyncService.batchInsertCameraBase(cameraBaseList);
            //3. 转换数据为CameraPlatform集合
            List<CameraPlatform> cameraPlatformList = new ArrayList<>();
            if (isSuccess) {
                cameraPlatformList = convertCameraPlatform(cameraBaseList);
            }
            return hkSyncService.batchInsertCameraPlatform(cameraPlatformList);
        }
        //没有需要同步的数据直接返回成功
        return true;
    }

    private List<CameraPlatform> convertToCameraPlatform(List<CameraChannel> channelList) {
        List<CameraPlatform> cameraPlatformList = new ArrayList<>();
        for (CameraChannel channel : channelList) {
            CameraPlatform cameraPlatform = new CameraPlatform();
            cameraPlatform.setOwnId(channel.getId());
            cameraPlatform.setType(2);
            cameraPlatform.setPlatformName("HK");
            cameraPlatform.setPlatformChannelCode(channel.getPlatformChannelCode());
            cameraPlatform.setIsSync(1);
            cameraPlatform.setStatus(1);
            cameraPlatformList.add(cameraPlatform);
        }
        return cameraPlatformList;
    }


    private List<CameraPlatform> convertCameraPlatform(List<CameraBase> cameraBaseList) {
        List<CameraPlatform> cameraPlatformList = new ArrayList<>();
        for (CameraBase cameraBase : cameraBaseList) {
            CameraPlatform cameraPlatform = new CameraPlatform();
            //设备ID
            cameraPlatform.setOwnId(cameraBase.getId());
            //类型
            cameraPlatform.setType(1);
            //第三方平台
            cameraPlatform.setPlatformName("HK");
            //区域编码
            cameraPlatform.setPlatformDomainCode(cameraBase.getRegionIndexCode());
            //设备编码
            cameraPlatform.setPlatformCameraCode(cameraBase.getEncodeDevIndexCode());
            //设置状态
            cameraPlatform.setStatus(1);
            //第三方平台是否同步;1已同步 0未同步
            cameraPlatform.setIsSync(1);
            cameraPlatformList.add(cameraPlatform);
        }
        return cameraPlatformList;
    }

    private Integer findByregionIndexCode(List<DistrictFeignDto> districtFeignDTOList, String regionIndexCode) {
        List<DistrictFeignDto> list = districtFeignDTOList
                .stream()
                .filter(item -> regionIndexCode.equals(item.getPlatformId()))
                .collect(Collectors.toList());
        if (list.size() > 0) {
            return list.get(0).getId();
        }
        return null;
    }

    private List<CameraBase> convertData(List<RegionDeviceDto> regionDeviceDtoList, List<DistrictFeignDto> districtFeignDTOList) {
        List<CameraBase> cameraBaseList = new ArrayList<>();
        for (RegionDeviceDto regionDeviceDTO : regionDeviceDtoList) {
            CameraBase cameraBase = new CameraBase();
            //设备名称
            cameraBase.setName(regionDeviceDTO.getName());
            //区域ID
            cameraBase.setDistrictId(findByregionIndexCode(districtFeignDTOList, regionDeviceDTO.getRegionIndexCode()));
            //设备编码
            cameraBase.setCode(UUID.fastUUID().toString());
            //接入协议
            if (EncodeDeviceProtocol.hiksdk_net.getProtocolName().equals(regionDeviceDTO.getTreatyType())) {
                cameraBase.setProtocolType(EncodeDeviceProtocol.hiksdk_net.getProtocol());
            } else if (EncodeDeviceProtocol.gb_reg.getProtocolName().equals(regionDeviceDTO.getTreatyType())) {
                cameraBase.setProtocolType(EncodeDeviceProtocol.gb_reg.getProtocol());
            }
            //ip地址
            cameraBase.setIp(regionDeviceDTO.getIp());
            //端口
            cameraBase.setPort(Integer.parseInt(regionDeviceDTO.getPort()));
            //设备账号
            cameraBase.setUserName(AesUtils.aesEnCode(regionDeviceDTO.getUserName()));
            //登录密码
            if (regionDeviceDTO.getPassword() != null) {
                cameraBase.setPassword(AesUtils.aesEnCode(regionDeviceDTO.getPassword()));
            }
            //设备厂商
            cameraBase.setCameraVendor(regionDeviceDTO.getManufacturer());
            //设备序列号
            cameraBase.setCameraSn(regionDeviceDTO.getDevSerialNum());
            //设备能力集
            cameraBase.setAbility(regionDeviceDTO.getCapability());
            //设备状态
            cameraBase.setStatus(1);
            //设备唯一编码
            cameraBase.setEncodeDevIndexCode(regionDeviceDTO.getIndexCode());
            //设别区域信息
            cameraBase.setRegionIndexCode(regionDeviceDTO.getRegionIndexCode());
            //加入集合
            cameraBaseList.add(cameraBase);
        }
        return cameraBaseList;
    }

    private boolean selectHkEncode(String data) {
        List<DistrictFeignDto> feignDTOList = JSONObject.parseObject(data, new TypeReference<List<DistrictFeignDto>>() {
        });
        //远程调用查询海康数据
        List<RegionDeviceDto> hkData = new ArrayList<>();
        for (DistrictFeignDto districtFeignDTO : feignDTOList) {
            String platformId = districtFeignDTO.getPlatformId();
            if (!StrUtil.isEmptyIfStr(platformId)) {
                Root<RegionDeviceDto> root = cameraService
                        .regionEncodeDevice(new PageInfo(1L, 1000L, new String[]{"view"}, platformId));
                if (root != null) {
                    hkData.addAll(root.getList());
                }
            }
        }
        Map<String, String> map = new HashMap<>(20);
        map.put("hkData", JSON.toJSONString(hkData));
        map.put("area", JSON.toJSONString(feignDTOList));
//        KafkaOpMsg kafkaOpMsg = new KafkaOpMsg();
//        kafkaOpMsg.setOpType("INSERT_OWN_CAMERA");
//        kafkaOpMsg.setData(JSON.toJSONString(map));
//        kafkaProducer.send(MediaMqConstants.CAMERA_TOPIC, JSON.toJSONString(kafkaOpMsg), true);
        return true;
    }
}



